package com.bujunjie.study.realtime.dws.app;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bujunjie.study.realtime.common.base.BaseApp;
import com.bujunjie.study.realtime.common.constant.FlinkConstant;
import com.bujunjie.study.realtime.common.function.DimAsyncFunction;
import com.bujunjie.study.realtime.common.function.Object2JsonStringFunction;
import com.bujunjie.study.realtime.common.util.FlinkSinkUtil;
import com.bujunjie.study.realtime.dws.app.model.sink.TradeProvinceOrderModel;
import com.bujunjie.study.realtime.dws.app.model.vo.TradeOrderDetailVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * <p>dws 层省份数据读取</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/18 15:39
 */
@Slf4j
public class DwsTradeProvinceOrderWindow extends BaseApp {
    public static void main(String[] args) throws Exception {
        new DwsTradeProvinceOrderWindow().start(
                10020,
                4,
                "dws_trade_province_order_window",
                FlinkConstant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<TradeOrderDetailVO> filterDS = kafkaStrDS.process(new ProcessFunction<String, TradeOrderDetailVO>() {
            @Override
            public void processElement(String value, ProcessFunction<String, TradeOrderDetailVO>.Context ctx, Collector<TradeOrderDetailVO> out) throws Exception {
                if (StrUtil.isNotBlank(value)) {
                    TradeOrderDetailVO tradeOrderDetailVO = JSON.parseObject(value, TradeOrderDetailVO.class);
                    out.collect(tradeOrderDetailVO);
                }

            }
        });
        KeyedStream<TradeOrderDetailVO, String> keyByDS = filterDS.keyBy(TradeOrderDetailVO::getId);

        //  数据去重，防止同一个订单id的值重复发放
        SingleOutputStreamOperator<TradeOrderDetailVO> distinctDS = keyByDS.process(new KeyedProcessFunction<String, TradeOrderDetailVO, TradeOrderDetailVO>() {

            /**
             *  状态声明
             */
            private ValueState<TradeOrderDetailVO> tradeOrderDetailState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<TradeOrderDetailVO> valueStateDescriptor = new ValueStateDescriptor<>("tradeOrderDetailState", TradeOrderDetailVO.class);
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.seconds(10)).build());
                tradeOrderDetailState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(TradeOrderDetailVO tradeOrderDetail, KeyedProcessFunction<String, TradeOrderDetailVO, TradeOrderDetailVO>.Context ctx, Collector<TradeOrderDetailVO> out) throws Exception {
                TradeOrderDetailVO value = tradeOrderDetailState.value();
                if (ObjectUtil.isNotNull(value)) {
                    //  代表当前订单已经存在
                    tradeOrderDetail.setSplitActivityAmount("-" + value.getSplitActivityAmount());
                    out.collect(value);
                }
                out.collect(tradeOrderDetail);
                //  更新此时的状态
                tradeOrderDetailState.update(tradeOrderDetail);
            }
        });

        // 指定流的水印
        SingleOutputStreamOperator<TradeOrderDetailVO> withWatermarksDS = distinctDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeOrderDetailVO>forBoundedOutOfOrderness(Duration.ofSeconds(2 * 60))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TradeOrderDetailVO>() {
                            @Override
                            public long extractTimestamp(TradeOrderDetailVO element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );

        // 按照省进行分组
        KeyedStream<TradeProvinceOrderModel, String> provinceDS = withWatermarksDS
                .map(new MapFunction<TradeOrderDetailVO, TradeProvinceOrderModel>() {
                    @Override
                    public TradeProvinceOrderModel map(TradeOrderDetailVO value) throws Exception {
                        TradeProvinceOrderModel tradeProvinceOrderModel = BeanUtil.copyProperties(value, TradeProvinceOrderModel.class);
                        tradeProvinceOrderModel.setOrderAmount(NumberUtil.toBigDecimal(value.getSplitTotalAmount()));
                        tradeProvinceOrderModel.setOrderIdSet(CollUtil.set(false, value.getOrderId()));
                        return tradeProvinceOrderModel;
                    }
                })
                .keyBy(TradeProvinceOrderModel::getProvinceId);

        //  10s 开窗进行轻度聚合
        WindowedStream<TradeProvinceOrderModel, String, TimeWindow> windowDS = provinceDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<TradeProvinceOrderModel> reduceDS = windowDS.reduce(new ReduceFunction<TradeProvinceOrderModel>() {
            @Override
            public TradeProvinceOrderModel reduce(TradeProvinceOrderModel value1, TradeProvinceOrderModel value2) throws Exception {
                value1.setOrderAmount(NumberUtil.add(value1.getOrderAmount(), value2.getOrderAmount()));
                value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                return value1;
            }
        }, new WindowFunction<TradeProvinceOrderModel, TradeProvinceOrderModel, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<TradeProvinceOrderModel> input, Collector<TradeProvinceOrderModel> out) throws Exception {
                TradeProvinceOrderModel model = input.iterator().next();
                model.setStartTime(DateUtil.format(new Date(window.getStart()), DatePattern.NORM_DATETIME_PATTERN));
                model.setEndTime(DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATETIME_PATTERN));
                model.setCurrentDate(DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATE_PATTERN));
                model.setOrderCount((long) model.getOrderIdSet().size());
                // 收集信息
                out.collect(model);
            }
        });

        SingleOutputStreamOperator<TradeProvinceOrderModel> provinceNameDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<TradeProvinceOrderModel>() {
            @Override
            public void addDims(TradeProvinceOrderModel provinceOrderModel, JSONObject dimJsonObj) {
                provinceOrderModel.setProvinceId(dimJsonObj.getString("name"));
            }

            @Override
            public String getTableName() {
                return "dim_base_province";
            }

            @Override
            public String getRowKey(TradeProvinceOrderModel tradeProvinceOrderModel) {
                return tradeProvinceOrderModel.getProvinceId();
            }
        }, 60, TimeUnit.SECONDS);

        provinceNameDS
                .map(new Object2JsonStringFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_province_order_window"));
    }
}
